package com.bw.gmall.realtime.app.dwd;

/*
t1 加购: 加购表--action
insert
update (new.num > old.num) new.num-old.num
维度退化
t0 flink jdbc_connector  缓存 100 缓存时间
t1 look up join

数据流向

1.业务库  2.flinkcdc 事实采集 maxwell

 */

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd_Test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
//        tenv.executeSql("select proctime()").print();

        //连接kafka topic_db
      tenv.executeSql(MyKafkaUtil.getTopicDb("cart_add_0625"));
      //查询结果



     //获取加购临时表
        Table table = tenv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['cart_price'] cart_price ,\n" +
                "cast(`data`['sku_num'] as bigint)-cast(`old`['sku_num'] as bigint) sku_num,\n" +
                "`data`['img_url'] img_url ,\n" +
                "`data`['sku_name'] sku_name,\n" +
                "`data`['is_checked'] is_checked,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['is_ordered'] is_ordered,\n" +
                "`data`['order_time'] order_time,\n" +
                "`data`['source_type'] source_type,\n" +
                "`data`['source_id'] source_id,\n" +
                "pt \n" +
                " from topic_db where `table`='cart_info' and (`type`='insert' or (`type`='update' and (`data`['sku_num'] > `old`['sku_num'])))");
        tenv.createTemporaryView("cart_info_e",table);

        //获取字典表
//      tenv.executeSql("CREATE TABLE base_dic (\n" +
//                "  dic_code String,\n" +
//                "  dic_name STRING,\n" +
//                "  parent_code String,\n" +
//                "  create_time String,\n" +
//                "  operate_time String,\n" +
//                "  PRIMARY KEY (dic_code) NOT ENFORCED\n" +
//                ") WITH (\n" +
//                "   'connector' = 'jdbc',\n" +
//                "   'username' = 'root',\n" +
//                "   'password' = 'root',\n" +
//                "   'url' = 'jdbc:mysql://hadoop102:3306/gmall',\n" +
//                "   'table-name' = 'base_dic'\n" +
//                ")\n"
//        );
//        tenv.executeSql("select * from base_dic").print();
        //获取字典表
        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());
        //字典表和加购表进行关联
        //获取加购临时表
        Table table1 = tenv.sqlQuery(" select\n" +
                " cart_info_e.id,\n" +
                " cart_info_e.user_id,\n" +
                " cart_info_e.sku_id,\n" +
                " cart_info_e.cart_price ,\n" +
                " cart_info_e.sku_num,\n" +
                " cart_info_e.img_url ,\n" +
                " cart_info_e.sku_name,\n" +
                " cart_info_e.is_checked,\n" +
                " cart_info_e.create_time,\n" +
                " cart_info_e.operate_time,\n" +
                " cart_info_e.is_ordered,\n" +
                " cart_info_e.order_time,\n" +
                " cart_info_e.source_type source_type_id,\n" +
                " bd.dic_name source_type_name,\n" +
                " cart_info_e.source_id\n" +
                " from cart_info_e join base_dic FOR SYSTEM_TIME AS OF cart_info_e.pt as bd" +
                " on cart_info_e.source_type=bd.dic_code");
        tenv.createTemporaryView("cart_info_basic",table1);

       tenv.executeSql("CREATE TABLE dwd_cart_add (\n" +
               "  `id` String,\n" +
               "  `user_id` String,\n" +
               "  `sku_id` String,\n" +
               "  `cart_price` String,\n" +
               "  `sku_num` BIGINT,\n" +
               "  `img_url` String,\n" +
               "  `sku_name` String,\n" +
               "  `is_checked` String,\n" +
               "  `create_time` String,\n" +
               "  `operate_time` String,\n" +
               "  `is_ordered` String,\n" +
               "  `order_time` String,\n" +
               "  `source_type_id` String,\n" +
               "  `source_type_name` String,\n" +
               "  `source_id` String\n" +
               ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_cart_add"));
       tenv.executeSql("insert into dwd_cart_add select * from cart_info_basic");
       tenv.executeSql("select * from dwd_cart_add").print();








    }
}
